package flink.windowCount;


import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.configuration.Configuration;

/**
 * 累加器
 */
public class AccumulatorApp {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        DataSource<String> data = env.fromElements("a", "b", "c", "d");
        DataSet<String> result = data.map(new RichMapFunction<String, String>() {

            //1.创建累加器
            private IntCounter numlines = new IntCounter();

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                //注册累加器
                getRuntimeContext().addAccumulator("num", this.numlines);
            }

            @Override
            public String map(String s) throws Exception {
                //如果并行度为1,使用普通的累加求和，结果正确，但是设置多个并行度，则普通的累加器求和结果不对
                //3.使用累加器
                this.numlines.add(1);
                System.out.println(getRuntimeContext().getAccumulator("num"));
                return s;
            }
        });
        result.print();

    }
}
